# 2.2 CrewAI 入门

> **第 2 周 · 第 2 课 · CrewAI 入门 · 预计时长: 45 分钟**

---

## 学习目标

- 掌握 CrewAI 的安装和初始化
- 理解 CrewAI 的三个核心概念：Agent、Task、Crew
- 能用 CrewAI 搭建最简单的多 Agent 协作系统
- 学会给 Crew 添加更多任务，实现流水线流程

---

## CrewAI 是什么？

CrewAI 是一个 Python 框架，专门用于构建多 Agent 协作系统。它把我们在 2.1 学到的概念（角色、任务、流程）封装成了简洁的 API。

**为什么选 CrewAI 而不是自己写？**

| 自己写 (纯 OpenAI SDK) | CrewAI |
|----------------------|--------|
| 手动管理 Agent 之间的数据传递 | 自动处理上下文传递 |
| 需要自己实现流程编排 | 内置 Sequential / Hierarchical 流程 |
| 角色定义是松散字典 | 类型安全的 Agent 类 |
| 输出格式不统一 | 支持 Pydantic / JSON 结构化输出 |

---

## 安装 CrewAI

### 前置条件

确保你已安装 Python 3.11+ 和 [uv](https://docs.astral.sh/uv/)（快速 Python 包管理器）。

如果没有 uv：

```bash
# macOS / Linux
curl -LsSf https://astral.sh/uv/install.sh | sh

# Windows (PowerShell)
powershell -c "irm https://astral.sh/uv/install.ps1 | iex"
```

### 创建项目并安装

```bash
# 创建项目目录
mkdir crewai-tutorial && cd crewai-tutorial

# 用 uv 初始化项目
uv init

# 安装 crewai（包含 openai 依赖）
uv add "crewai[tools]"

# 激活虚拟环境
source .venv/bin/activate      # macOS/Linux
# 或
.venv\Scripts\activate         # Windows
```

> **提示：** 安装 `crewai[tools]`（带方括号）会包含常用的工具包，如网页搜索、文件读写等。

### 设置 API Key

```bash
# macOS / Linux
export OPENAI_API_KEY="your-api-key-here"

# Windows (PowerShell)
$env:OPENAI_API_KEY="your-api-key-here"
```

---

## CrewAI 核心概念

CrewAI 有三个核心类，对应我们 2.1 学到的概念：

```
Agent  = 角色（谁来做）
Task   = 任务（做什么）
Crew   = 团队（怎么协作）
```

### Agent — 角色定义

```python
from crewai import Agent

researcher = Agent(
    role="市场研究员",                          # 角色名称
    goal="分析市场趋势，找出关键数据和洞察",       # 这个 Agent 要达成什么
    backstory=(                                 # 专业背景和人设
        "你是一位有 8 年经验的市场研究员，"
        "擅长从海量数据中提取关键趋势，"
        "你的分析报告被多家知名咨询公司引用。"
    ),
    verbose=True,                               # 输出详细日志
    allow_delegation=False,                     # 是否允许委托任务给其他 Agent
    # llm=...,                                  # 可选：指定使用的模型
)
```

**参数说明：**

| 参数 | 说明 | 默认值 |
|------|------|--------|
| `role` | 角色名称 | 必填 |
| `goal` | 角色目标 | 必填 |
| `backstory` | 背景故事 | 必填 |
| `verbose` | 是否输出详细日志 | `False` |
| `allow_delegation` | 是否可委托任务 | `False` |
| `llm` | 使用的语言模型 | 默认 `gpt-4o` |
| `tools` | 可用工具列表 | `[]` |
| `memory` | 是否启用记忆 | `False` |

### Task — 任务定义

```python
from crewai import Task

research_task = Task(
    description=(                                    # 任务描述
        "研究 2025 年 AI 健身 App 市场趋势，"
        "找出 Top 5 竞争对手、用户痛点、增长机会。"
    ),
    expected_output=(                                # 期望输出格式
        "一份市场分析报告，包含：竞争格局、用户痛点 Top 3、"
        "市场增长机会 Top 3，每个点附简要说明。"
    ),
    agent=researcher,                                # 指派给哪个 Agent
    # output_json=MarketReport,                     # 可选：输出为 Pydantic 对象
    # output_file="reports/market_analysis.md",     # 可选：保存到文件
)
```

### Crew — 团队协作

```python
from crewai import Crew, Process

crew = Crew(
    agents=[researcher],            # 参与的 Agent 列表
    tasks=[research_task],          # 要执行的任务列表
    process=Process.sequential,     # 协作流程：sequential 或 hierarchical
    verbose=True,                   # 显示详细执行日志
)

# 启动 Crew
result = crew.kickoff()
print(result)
```

**Process 类型：**

- `Process.sequential`：任务按顺序执行，上一个的输出自动作为下一个的上下文
- `Process.hierarchical`：Manager Agent 自动分配和协调任务

---

## 实战 1：最简单的 Crew

让我们从零搭建一个最简单的多 Agent Crew。

### 场景：写一条社交媒体帖子

```python
# 文件: simple_crew.py

from crewai import Agent, Task, Crew, Process

# ===== 第 1 步：定义 Agent =====

# Agent 1：创意策划
ideator = Agent(
    role="社交媒体创意策划",
    goal="为 AI 健身 App 想出一个吸引人的社交媒体帖子创意",
    backstory=(
        "你是社交媒体营销专家，曾为多个科技品牌策划过 viral 营销内容。"
        "你擅长用简洁有力的文案抓住用户注意力。"
    ),
    verbose=True,
)

# ===== 第 2 步：定义 Task =====

ideation_task = Task(
    description=(
        "为即将发布的 AI 健身 App 设计一个 Instagram 帖子。"
        "帖子需要：1) 吸引人的标题  2) 正文文案（100-150字）"
        " 3) 3-5 个相关 hashtag"
    ),
    expected_output="一个完整的 Instagram 帖子文案，包含标题、正文和 hashtag",
    agent=ideator,
)

# ===== 第 3 步：创建 Crew 并执行 =====

crew = Crew(
    agents=[ideator],
    tasks=[ideation_task],
    process=Process.sequential,
    verbose=True,
)

# 启动！
result = crew.kickoff()

print("\n" + "=" * 60)
print("最终结果:")
print("=" * 60)
print(result)
```

### 运行

```bash
uv run python simple_crew.py
```

### 典型输出

```
[2025-01-15 10:30:00] ## 开始执行任务: 为即将发布的 AI 健身 App 设计...

[2025-01-15 10:30:15] ## 思考: 我需要考虑目标受众...

[2025-01-15 10:30:25] ## 最终答案:
# Instagram 帖子

**标题:** 你的私人 AI 健身教练，终于来了

**正文:**
还在为找不到合适的训练计划发愁？AI 健身 App 帮你搞定！
只需告诉我你的目标、时间和可用器械，AI 会为你定制专属训练方案。
更棒的是，它会实时纠正你的动作，就像身边有个专业教练。
下周首发，早鸟用户享 5 折！评论区留下你的健身目标，
我会抽 10 位送出终身会员
**Hashtags:**
#AI健身 #智能健身 #私人教练 #健身App #健康科技

============================================================
最终结果:
============================================================
[上面的帖子内容]
```

> **注意：** 实际输出格式可能因版本和模型不同而有所差异，但核心流程一致。

---

## 实战 2：添加第二个任务 — 流水线

现在我们在创意策划之后，加一个 "文案润色" 任务。

```python
# 文件: pipeline_crew.py

from crewai import Agent, Task, Crew, Process

# ===== 定义两个 Agent =====

# Agent 1：创意策划
ideator = Agent(
    role="社交媒体创意策划",
    goal="为 AI 健身 App 想出一个吸引人的社交媒体帖子创意",
    backstory=(
        "你是社交媒体营销专家，曾为多个科技品牌策划过 viral 营销内容。"
    ),
    verbose=True,
)

# Agent 2：文案编辑
editor = Agent(
    role="资深文案编辑",
    goal="润色和优化社交媒体文案，使其更具吸引力和专业感",
    backstory=(
        "你有 10 年文案编辑经验，曾在 4A 广告公司工作。"
        "你擅长将普通文案升级为让人忍不住想点赞的内容。"
        "你的标准是：每句话都要有存在的理由。"
    ),
    verbose=True,
)

# ===== 定义两个 Task =====

ideation_task = Task(
    description=(
        "为即将发布的 AI 健身 App 设计一个 Instagram 帖子。"
        "包含：标题、正文（100-150字）、3-5 个 hashtag。"
    ),
    expected_output="一个 Instagram 帖子草稿，包含标题、正文和 hashtag",
    agent=ideator,
)

editing_task = Task(
    description=(
        "审核并润色以下社交媒体帖子草稿。"
        "你需要：1) 优化标题使其更抓人  2) 精简正文，删掉冗余句子"
        " 3) 检查 hashtag 的相关性和热度  4) 给出修改理由说明。"
        "注意：保持原文核心信息不变。"
    ),
    expected_output="润色后的帖子 + 修改说明（列出每个修改点及理由）",
    agent=editor,
    # 注意：这里没有 context 参数，CrewAI 在 sequential 模式下
    # 会自动将上一个任务的输出作为上下文传递给当前任务
)

# ===== 创建 Crew =====

crew = Crew(
    agents=[ideator, editor],
    tasks=[ideation_task, editing_task],
    process=Process.sequential,
    verbose=True,
)

# 执行流水线
result = crew.kickoff()

print("\n" + "=" * 60)
print("流水线最终输出:")
print("=" * 60)
print(result)
```

### 关键理解

在 `Process.sequential` 模式下：
1. CrewAI 先执行 `ideation_task`（Agent: ideator）
2. `ideation_task` 的输出**自动**成为 `editing_task` 的上下文
3. `editor` Agent 能看到前一个任务的结果并基于它工作

这就是多 Agent 流水线的核心机制 — **上下文自动传递**。

### 典型输出

```
============================================================
流水线最终输出:
============================================================
润色后的 Instagram 帖子：

**标题:** 健身不再靠毅力 — AI 教练懂你每一步

**正文:**
别再跟着千篇一律的教程练了。
AI 健身 App 根据你的体能、时间和目标，实时生成专属训练计划。
动作不规范？AI 实时纠正，效果堪比私教。
下周上线，早鸟 5 折。
告诉我你的健身目标，抽 10 位送终身会员

**Hashtags:** #AI健身 #智能训练 #健身黑科技 #私人教练平替 #自律不如选对工具

--- 修改说明 ---
1. 标题：从"私人教练"改为"AI 教练懂你每一步"，更口语化，制造好奇
2. 正文：从 120 字精简到 75 字，删除了冗余的过渡句，节奏更紧凑
3. Hashtag：将"智能健身"改为"智能训练"，避免与竞品重复；新增"自律不如选对工具"作为 viral tag
```

---

## 实战 3：一个 Agent 执行多个任务

有些场景下，同一个 Agent 可以做多个任务。

```python
# 文件: multi_task_crew.py

from crewai import Agent, Task, Crew, Process

# 只有一个 Agent
writer = Agent(
    role="技术文档写手",
    goal="撰写清晰、简洁、结构化的技术文档",
    backstory="你有 8 年技术写作经验，擅长把复杂概念讲得通俗易懂。",
    verbose=True,
)

# 但有两个任务 — 都由同一个 writer 完成
task_outline = Task(
    description="为 REST API 教程写一份大纲，包含 5 个章节标题和每章要点",
    expected_output="一份详细的教程大纲",
    agent=writer,
)

task_content = Task(
    description="基于上面的大纲，撰写第 1 章的完整内容，至少 500 字",
    expected_output="第 1 章的完整教程内容",
    agent=writer,
)

crew = Crew(
    agents=[writer],
    tasks=[task_outline, task_content],
    process=Process.sequential,
    verbose=True,
)

result = crew.kickoff()
print(result)
```

> 注意：虽然只有一个 Agent，但 `sequential` 流程保证了 `task_content` 能看到 `task_outline` 的输出。

---

## 使用自定义模型

如果你不想用 OpenAI，可以指定其他模型：

```python
from crewai import Agent, LLM

# 使用本地 Ollama 模型
local_llm = LLM(
    model="ollama/llama3.2",
    base_url="http://localhost:11434",
)

agent = Agent(
    role="数据分析师",
    goal="分析销售数据趋势",
    backstory="你是资深数据分析师...",
    llm=local_llm,   # 指定自定义模型
)
```

---

## 动手练习

### 练习 1：搭建一个 "旅行规划" Crew

**目标：** 创建一个由 2 个 Agent 组成的旅行规划系统。

- Agent 1（目的地研究员）：根据用户偏好推荐 3 个目的地
- Agent 2（行程规划师）：为选定的目的地规划 5 天行程

**要求：**
1. 使用 `Process.sequential`
2. Agent 2 的输入包含 Agent 1 的推荐结果
3. 每个 Agent 的 backstory 要具体

**参考代码框架：**

```python
from crewai import Agent, Task, Crew, Process

researcher = Agent(
    role=???,
    goal=???,
    backstory=???,
    verbose=True,
)

planner = Agent(
    role=???,
    goal=???,
    backstory=???,
    verbose=True,
)

research_task = Task(
    description=???,
    expected_output=???,
    agent=researcher,
)

planning_task = Task(
    description=???,
    expected_output=???,
    agent=planner,
)

crew = Crew(
    agents=[???],
    tasks=[???],
    process=Process.sequential,
    verbose=True,
)

result = crew.kickoff()
print(result)
```

### 练习 2：尝试添加第三个 Agent

在练习 1 的基础上，添加一个 "预算评估师" Agent，作为流水线的第三步。它的任务是根据行程规划，估算旅行总费用并给出省钱建议。

---

## 常见问题

### Q1: 任务之间的上下文是怎么传递的？

在 `Process.sequential` 模式下，CrewAI 自动将前一个任务的输出添加到下一个任务的上下文中。你不需要手动传递。

### Q2: 如何指定不同的模型？

在 Agent 中设置 `llm` 参数：
```python
agent = Agent(..., llm=LLM(model="gpt-4o-mini"))
```

### Q3: 一个 Agent 可以执行多个任务吗？

可以。只需在多个 Task 中引用同一个 Agent。

### Q4: verbose=True 输出太多日志了怎么办？

设置为 `False` 即可，只会输出最终结果。

---

## 总结

- **CrewAI 三要素**：Agent（角色）、Task（任务）、Crew（团队）
- **安装**：`uv add "crewai[tools]"`
- **Sequential 流程**：任务按顺序执行，上下文自动传递
- **一个 Agent 可做多个任务**，也可多个 Agent 各做一个任务
- **verbose=True** 帮助调试，`False` 用于生产环境

**下节课预告：** 我们将用 CrewAI 搭建一个完整的 AI 辩论系统 — 两个 Agent 针锋相对，第三个 Agent 当裁判打分！